package com.hmall.item.es;

import cn.hutool.json.JSONUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.hmall.common.utils.CollUtils;
import com.hmall.item.domain.po.Item;
import com.hmall.item.domain.po.ItemDoc;
import com.hmall.item.service.IItemService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author zzw
 * @date 2025/6/24
 * @description 测试数据批量导入
 */

@Slf4j
@Service
@RequiredArgsConstructor
public class ESDataBulk {

    private final IItemService itemService;
    private static ElasticsearchClient esClient;
    private static RestClient restClient;

    public static void connect(String host, int port) {
        restClient = RestClient.builder(new HttpHost(host, port)).build();

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());

        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper(objectMapper));

        esClient = new ElasticsearchClient(transport);
    }

    public void esBulk() throws IOException {
        // 初始化 currentPage，注意mp的分页从1开始
        int currentPage = 1;
        while (true) {
            // 取第一页1000条数据
            Page<Item> page = Page.of(currentPage, 1000);
            // 查询所有商品信息
            Page<Item> itemPage = itemService.page(
                    page, new LambdaQueryWrapper<>());
            // 获取商品集合
            List<Item> items = itemPage.getRecords();

            if (CollUtils.isEmpty(items)) {
                break;
            }

            List<ItemDoc> itemDocs = items.stream().map(item -> {
                ItemDoc itemDoc = new ItemDoc();

                itemDoc.setId(String.valueOf(item.getId()));

                // 拷贝基础属性
                BeanUtils.copyProperties(item, itemDoc);

                // 拷贝规格属性
                Set<Map.Entry<String, Object>> entries = JSONUtil.parseObj(item.getSpec()).entrySet();
                log.info("规格属性：{}", entries);

                List<ItemDoc.Spec> specs = new ArrayList<>();
                entries.forEach(entry -> {
                    ItemDoc.Spec spec = new ItemDoc.Spec();
                    spec.setName(entry.getKey());
                    spec.setValue(String.valueOf(entry.getValue()));
                    specs.add(spec);
                });
                itemDoc.setAttr_list(specs);

                // 添加自动补全字段
                itemDoc.setSuggestion(List.of(
                        item.getCategory(),
                        item.getBrand()
                                             ));

                return itemDoc;
            }).collect(Collectors.toList());


            // 批量添加请求
            BulkRequest.Builder br = new BulkRequest.Builder();
            itemDocs.forEach(itemDoc ->
                                     br.operations(op -> op
                                             .index(i -> i
                                                     .index("items")
                                                     .id(itemDoc.getId())
                                                     .document(itemDoc))));
            // 构建请求
            BulkRequest build = br.build();
            // 批量添加
            BulkResponse bulkResponse = esClient.bulk(build);
            // 遍历结果
            bulkResponse.items().forEach(item -> {
                if (item.result() != null) {
                    log.info("添加结果：{}", item.result());
                }
            });

            // 如果有错误
            if (bulkResponse.errors()) {
                log.error("批量添加失败");
                // 遍历错误
                bulkResponse.items().forEach(item -> {
                    if (item.error() != null) {
                        log.error("添加失败：{}", item.error().reason());
                    }
                });
            }

            // 下一页
            currentPage++;
        }
    }


    public static void close() throws IOException {
        restClient.close();
    }

}
